package com.nd.bdc.presto.websocket;

import com.nd.bdc.presto.service.PrestoWebSocketService.*;
import com.nd.bdc.presto.websocket.disruptor.*;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;


@Component
public class MessageDisruptorQueue {

    private Executor executor = Executors.newCachedThreadPool();

    // The factory for the event
    private MessageEventFactory factory = new MessageEventFactory();


    // Specify the size of the ring buffer, must be power of 2.
    private int bufferSize = 2 * 1024;

    // Construct the Disruptor
    private Disruptor<MessageEvent> disruptor = new Disruptor<>(factory, bufferSize, executor);;


    private static  RingBuffer<MessageEvent> ringBuffer;

    @Autowired
    MessageDisruptorQueue(MessageEventHandler eventHandler){
        disruptor.handleEventsWith(eventHandler);
        this.ringBuffer = disruptor.getRingBuffer();
        disruptor.start();
    }


    public static void publishEvent(String channel, MessageData messageData) {
        long sequence = ringBuffer.next();  // Grab the next sequence
        try {
            MessageEvent event = ringBuffer.get(sequence); // Get the entry in the Disruptor
            // for the sequence
            event.setMessageData(messageData);  // Fill with data
            event.setChannel(channel);
        } finally {
            ringBuffer.publish(sequence);
        }
    }


}
